﻿using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using System.Transactions;
using Common.Utilities;
using Common.Utilities.Dispatcher;
using Common.Utilities.TextMatch;

namespace Workflows.Components.TextMining.Ontologies
{
    /// <summary>
    /// 
    /// </summary>
    public class WordFreqWorker:IWorker 
    {
        #region Implementation of IWorker
        /// <summary></summary>
        public event WorkerStarted WorkerStartedEvent;
        /// <summary></summary>
        public event WorkerFinished WorkerFinishedEvent;
        /// <summary></summary>
        public event WorkerHasError WorkerErredEvent;
        /// <summary></summary>
        public event WorkerCancelled WorkerCancelledEvent;
        /// <summary></summary>
        public event WorkerStatusChanged WorkerStatusChangedEvent;

        private Thread workerThread;
        private WordFreqInstruction instruction;
        private WordFreqOutput output;
        private RunContext _Context;
        private Guid _TicketID;
        private string _WorkflowName;
        private string _ComponentName;

        /// <summary>
        /// 
        /// </summary>
        public string ComponentName
        {
            get { return _ComponentName; }
        }

        /// <summary>
        /// 
        /// </summary>
        public string WorkflowName
        {
            get { return _WorkflowName; }
        }

        /// <summary></summary>
        public Thread WorkerThread
        {
            get { return workerThread; }
        }
        /// <summary></summary>
        public InstructionBase Instruction
        {
            get { return instruction; }
        }
        /// <summary></summary>
        public ResultBase Output
        {
            get { return output; }
        }
        /// <summary></summary>
        public Type ComponentType
        {
            get { return typeof(WordFreqComponent); }
        }
        /// <summary></summary>
        public RunContext Context
        {
            get { return _Context; }
            set { _Context = value; }
        }
        /// <summary></summary>
        public void SetInstruction(Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._TicketID = ticketGuid;
            this.instruction = (WordFreqInstruction)request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }
        /// <summary></summary>
        public void SetInstruction(RunContext context, Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._TicketID = ticketGuid;
            this.instruction = (WordFreqInstruction)request;
            this._Context = context;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }
        /// <summary></summary>
        public void DoWork()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this.instruction);
            }
            this.workerThread = Thread.CurrentThread;
            this.Index();
        }
        /// <summary></summary>
        public void DoWorkAsync()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this.instruction);
            }
            this.workerThread = new Thread(new ThreadStart(this.Index));
            this.workerThread.Start();
        }
        /// <summary></summary>
        public void CancelWork()
        {
            if (this.workerThread != null && this.workerThread.IsAlive)
            {
                this.workerThread.Abort();
                this.workerThread.Join();

                if (this.WorkerCancelledEvent != null)
                {
                    this.WorkerCancelledEvent(this._TicketID, this.GetType(), this.instruction);
                }
            }
        }

        #endregion

        #region work
        private void Index()
        {
            SqlConnection srcConn=new SqlConnection(this.instruction.ConnStr);
            SqlConnection tgtConn=new SqlConnection(this.instruction.OutputConnStr);
            try
            {
                if(!this.CheckTermsTable())
                    throw new Exception("Unable to create term table or find it");
                if(!this.CheckDocTermTable())
                    throw new Exception("Unable to create doc term table or find it");
                bool indexNeighbors = false;
                if(!string.IsNullOrEmpty(this.instruction.NeighborTable))
                {
                    indexNeighbors = true;
                    if(!this.CheckNeighborTable())
                        throw new Exception("Unable to create neighbor table or find it");
                }
                bool trackTrend = false;
                if(!string.IsNullOrEmpty(this.instruction.TrendTable))
                {
                    trackTrend = true;
                    if(!this.CheckTrendTable())
                        throw new Exception("Unable to create trend table or find it");
                }

                int totalRecords = DbUtil.GetTableRecordCount(
                    DataProviderType.MSSQL, instruction.ConnStr, instruction.SQL);

                string spInsert;
                if(!this.AddTermProcSP(out spInsert))
                    throw new Exception("Unable to create procedure");
                string spLookupTerm;
                if(!this.LookupTermSP(out spLookupTerm))
                    throw new Exception("Unable to create procedure");
                
                tgtConn.Open();
                SqlCommand tgtCmd = tgtConn.CreateCommand();
                tgtCmd.CommandType = CommandType.StoredProcedure;

                srcConn.Open();
                SqlCommand srcCmd = srcConn.CreateCommand();
                srcCmd.CommandType = CommandType.Text;
                srcCmd.CommandText = instruction.SQL;
                SqlDataReader reader = srcCmd.ExecuteReader();
                int docCount = 0;
                int corpusSize = 0;
                int termCount = 0;
                // PorterStemmer stemmer=new PorterStemmer();
                StopWordsUtil stopWordsUtil=new StopWordsUtil(StopWordsUtil.StopWordList.PubmedStopWordList);
                List<string> stopWords = stopWordsUtil.StopWords;
                while(reader.Read())
                {
                    if(reader[instruction.TextField]!=DBNull.Value && 
                        reader[instruction.DocIDField]!=DBNull.Value)
                    {
                        docCount++;
                        Dictionary<string,int> wordTerms=new Dictionary<string, int>();
                        Dictionary<int,int> termFreqs=new Dictionary<int, int>();
                        string text = reader[instruction.TextField].ToString();
                        int docID = int.Parse(reader[instruction.DocIDField].ToString());
                        int? year = null;
                        if(!string.IsNullOrEmpty(instruction.YearField) &&  
                            reader[instruction.YearField]!=DBNull.Value)
                        {
                            year = int.Parse(reader[instruction.YearField].ToString());
                        }
                        try
                        {
                            Dictionary<string, int> wordFreqs = this.TokenizeText(text);
                            foreach (string word in wordFreqs.Keys)
                            {
                                corpusSize += wordFreqs[word];
                                bool isNewTerm = false;
                                int termID = this.AddWordFrequency(tgtCmd, spLookupTerm, spInsert, word, 0, ref isNewTerm);
                                if (isNewTerm)
                                {
                                    termCount++;
                                }
                                this.AddDocTermMappings(tgtCmd, docID, termID, wordFreqs[word]);
                                if (!wordTerms.ContainsKey(word))
                                    wordTerms.Add(word, termID);
                                if (!termFreqs.ContainsKey(termID))
                                    termFreqs.Add(termID, wordFreqs[word]);
                            }

                            if (indexNeighbors)
                                this.ProcessSentences(tgtCmd, text, stopWords, termFreqs, wordTerms);
                            if (trackTrend && year.HasValue)
                                this.AddTermToTrend(tgtCmd, year.Value, termFreqs);
                        }
                        catch (Exception ex1)
                        {
                            if (this.WorkerErredEvent != null)
                                this.WorkerErredEvent(this._TicketID, this.GetType(), this.instruction,
                                                      ex1.ToString(), false);
                            if (tgtConn.State != ConnectionState.Open)
                                tgtConn.Open();
                        }
                    }

                    if(this.WorkerStatusChangedEvent !=null && docCount % 100==0)
                    {
                        string msg = string.Format("Processed {0} of {1}..", docCount, totalRecords);
                        int pct = docCount*100/totalRecords;
                        this.WorkerStatusChangedEvent(
                            this._TicketID, this.GetType(), this.instruction, this._WorkflowName,
                            this._ComponentName, msg, pct);
                    }
                }
                reader.Close();

                if(this.WorkerStatusChangedEvent !=null && docCount % 100 ==0)
                {
                    string msg = string.Format("Processed {0} of {1} records...", docCount, totalRecords);
                    int pct = docCount*100/totalRecords;
                    this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this.instruction,
                                                  this._WorkflowName, this._ComponentName, msg, pct);
                }

                output = new WordFreqOutput(instruction.OwnerComponentID, this._TicketID, termCount, docCount,
                                            corpusSize);
                if(this.WorkerFinishedEvent !=null)
                {
                    this.WorkerFinishedEvent(this._TicketID, this.GetType(), instruction, output);
                }
            }
            catch(Exception ex)
            {
                if (this.WorkerErredEvent != null)
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this.instruction, ex.ToString(), true);
            }
            finally
            {
                srcConn.Close();
                tgtConn.Close();
            }
        }
        #endregion

        #region target tables
        private bool CheckTermsTable()
        {
            if(DbUtil.IsTableExist(DataProviderType.MSSQL, this.instruction.OutputConnStr, this.instruction.TermTable))
                return true;

            List<TableColumn> cols=new List<TableColumn>();
            cols.Add(new TableColumn()
                         {
                             ColumnName = this.instruction.TermIDField,
                             DbType = SqlDbType.Int,
                             IsPK=true
                         });
            cols.Add(new TableColumn()
                         {
                             ColumnName=this.instruction.PhraseField,
                             DbType=SqlDbType.VarChar,
                             AllowNull=false,
                             Size=50
                         });
            cols.Add(new TableColumn()
                         {
                             ColumnName=this.instruction.RootTermIDField,
                             DbType=SqlDbType.Int,
                             AllowNull=true,
                         });
            TableHelper.CreateTable(instruction.OutputConnStr,instruction.TermTable,cols.ToArray(), false);
            return true;
        }

        private bool CheckDocTermTable()
        {
            if (DbUtil.IsTableExist(
                DataProviderType.MSSQL, this.instruction.OutputConnStr, 
                this.instruction.DocTermTable))
                return true;

            List<TableColumn> cols = new List<TableColumn>();
            cols.Add(new TableColumn()
            {
                ColumnName = this.instruction.TermIDField2,
                DbType = SqlDbType.Int,
                AllowNull=false
            });
            cols.Add(new TableColumn()
            {
                ColumnName = this.instruction.DocIDField2,
                DbType = SqlDbType.Int,
                AllowNull = false
            });
            cols.Add(new TableColumn()
            {
                ColumnName = this.instruction.TermFreqField,
                DbType = SqlDbType.Int,
                AllowNull = false,
            });
            TableHelper.CreateTable(
                instruction.OutputConnStr, instruction.DocTermTable, 
                cols.ToArray(), false);
            return true;
        }

        private bool CheckNeighborTable()
        {
            if (DbUtil.IsTableExist(
                DataProviderType.MSSQL, this.instruction.OutputConnStr,
                this.instruction.NeighborTable))
                return true;

            List<TableColumn> cols = new List<TableColumn>();
            cols.Add(new TableColumn()
            {
                ColumnName = this.instruction.LeftTermIDField,
                DbType = SqlDbType.Int,
                AllowNull = false
            });
            cols.Add(new TableColumn()
            {
                ColumnName = this.instruction.RightTermIDField,
                DbType = SqlDbType.Int,
                AllowNull = false
            });
            cols.Add(new TableColumn()
            {
                ColumnName = this.instruction.NeighborTotalOccuranceField,
                DbType = SqlDbType.Int,
                AllowNull = false,
            });
            cols.Add(new TableColumn()
            {
                ColumnName = this.instruction.DocContainingNeighborField,
                DbType = SqlDbType.Int,
                AllowNull = false,
            });
            TableHelper.CreateTable(
                instruction.OutputConnStr, instruction.NeighborTable,
                cols.ToArray(), false);
            return true;
        }

        private bool CheckTrendTable()
        {
            if (DbUtil.IsTableExist(
                DataProviderType.MSSQL, this.instruction.OutputConnStr,
                this.instruction.TrendTable))
                return true;

            List<TableColumn> cols = new List<TableColumn>();
            cols.Add(new TableColumn()
            {
                ColumnName = this.instruction.TermIDField3,
                DbType = SqlDbType.Int,
                AllowNull = false
            });
            cols.Add(new TableColumn()
            {
                ColumnName = this.instruction.TermYearField,
                DbType = SqlDbType.Int,
                AllowNull = false
            });
            cols.Add(new TableColumn()
            {
                ColumnName = this.instruction.SumTermFreqField,
                DbType = SqlDbType.Int,
                AllowNull = false,
            });
            cols.Add(new TableColumn()
            {
                ColumnName = this.instruction.SumDocFreqField,
                DbType = SqlDbType.Int,
                AllowNull = false,
            });
            TableHelper.CreateTable(
                instruction.OutputConnStr, instruction.TrendTable,
                cols.ToArray(), false);
            return true;
        }
        #endregion

        #region stored procs
        private bool AddTermProcSP(out string spName)
        {
            spName = "AddTerm_" + instruction.TermTable;
            string spInsertBody = @"create procedure {0} (
@{2} int output,
@{3} int,
@{4} varchar(50)) as 
begin
	set @TermID=(select max([{2}]) from dbo.[{1}]);
	if(@TermID is null)
		set @TermID=1;
	else 
		set @TermID=@TermID+1;
	insert into [{1}] ([{2}],[{3}],[{4}]) values (
    @{2},
    @{3},
    @{4});
end";
            spInsertBody = string.Format(spInsertBody, spName, instruction.TermTable,
                                         instruction.TermIDField, instruction.RootTermIDField,
                                         instruction.PhraseField);
            return DbUtil.NewStoredProc(this.instruction.OutputConnStr, spName, spInsertBody);
        }

        private bool LookupTermSP(out string spName)
        {
            spName = "LookupTerm";
            string spBody = @"create procedure {0} 
(@{2} int output,
@{3} varchar(50)) as 
begin
	set @{2}=(select top 1 [{2}] from [{1}] where [{3}]=@{3});
    if (@{2} is null) set @{2}=0;
end";
            spBody = string.Format(
                spBody, spName, instruction.TermTable,
                instruction.TermIDField, instruction.PhraseField);
            return DbUtil.NewStoredProc(this.instruction.OutputConnStr, spName, spBody);
        }
        #endregion

        #region terms
        private int AddWordFrequency(
            SqlCommand cmd, string spLookup, string spInsert, 
            string word, int rootTermID, ref bool isNewTerm)
        {
            using (TransactionScope scope = new TransactionScope())
            {
                cmd.CommandType = CommandType.StoredProcedure;
                cmd.CommandText = spLookup;
                cmd.Parameters.Clear();
                SqlParameter termIDParameter1 = new SqlParameter()
                                                    {
                                                        ParameterName = instruction.TermIDField,
                                                        DbType = DbType.Int32,
                                                        Direction = ParameterDirection.Output
                                                    };
                cmd.Parameters.Add(termIDParameter1);
                cmd.Parameters.AddWithValue(this.instruction.PhraseField, word);
                cmd.ExecuteNonQuery();
                int termID = int.Parse(termIDParameter1.Value.ToString());
                if (termID <= 0)
                {
                    cmd.CommandText = spInsert;
                    cmd.CommandType = CommandType.StoredProcedure;
                    cmd.Parameters.Clear();
                    SqlParameter termIDParameter = new SqlParameter()
                                                       {
                                                           DbType = DbType.Int32,
                                                           ParameterName = instruction.TermIDField,
                                                           Direction = ParameterDirection.Output
                                                       };
                    cmd.Parameters.Add(termIDParameter);
                    cmd.Parameters.AddWithValue(instruction.RootTermIDField, rootTermID);
                    cmd.Parameters.AddWithValue(instruction.PhraseField, word);
                    cmd.ExecuteNonQuery();
                    termID = int.Parse(termIDParameter.Value.ToString());
                    isNewTerm = true;
                }
                scope.Complete();
                return termID;
            }
        }

        private Dictionary<string,int> TokenizeText(string text)
        {
            Dictionary<string,int> wordFreqs=new Dictionary<string, int>();
            Tokeniser tokeniser=new Tokeniser();
            string[] words = tokeniser.Partition(text.ToLower());
            Regex wordRegex = new Regex("^[^a-zA-Z]*([a-zA-Z]+)[a-zA-Z]*$");
            foreach(string word in words)
            {
                if(wordRegex.IsMatch(word) && word.Trim().Length>=2)
                {
                    if (wordFreqs.ContainsKey(word))
                        wordFreqs[word] = wordFreqs[word] + 1;
                    else 
                        wordFreqs.Add(word,1);
                }
            }
            return wordFreqs;
        }
        #endregion

        #region doc-term mappings
        private void AddDocTermMappings(SqlCommand cmd, 
            int docID, int termID, int termFreq)
        {
            cmd.CommandType = CommandType.Text;
            string sql =
                @"
if not exists (select * from [{0}] where [{1}]={4} and [{2}]={5})
    insert into [{0}] ([{1}],[{2}],[{3}]) values ({4},{5},{6})";
            cmd.CommandText = string.Format(sql,
                instruction.DocTermTable, instruction.TermIDField2,
                instruction.DocIDField2, instruction.TermFreqField,
                termID, docID, termFreq);
            cmd.ExecuteNonQuery();
        }
        #endregion

        #region neighbors
        private void ProcessSentences(
            SqlCommand cmd, string docText,
            List<string> stopWords,
            Dictionary<int,int> termFreqInDoc,
            Dictionary<string,int> wordTermIDs)
        {
            int totalWordsInDoc = 0;
            foreach (int termID in termFreqInDoc.Keys)
                totalWordsInDoc += termFreqInDoc[termID];

            List<string> sentences = SentenceBoundaryUtil.GetSentences(docText);
            Dictionary<int,Dictionary<int,double>> neighborProbs=new Dictionary<int, Dictionary<int, double>>();
            Dictionary<int, Dictionary<int, int>> neighborFreqs = new Dictionary<int, Dictionary<int, int>>();
            foreach(string sentence in sentences)
            {
                string[] wordsInSentence = sentence.ToLower().Split(new char[] { ',', '.', ' ' });
                if(wordsInSentence.Length<2)
                    continue;
                double sentenceCoverage = (double) sentence.Length/docText.Length;
                int lastNonStopTermID = 0;
                double lastTermFreq = 0;
                foreach (string word in wordsInSentence)
                {
                    if (wordTermIDs.ContainsKey(word) && !stopWords.Contains(word))
                    {
                        if (lastNonStopTermID == 0)
                        {
                            lastNonStopTermID = wordTermIDs[word];
                            lastTermFreq = (double) termFreqInDoc[lastNonStopTermID]/totalWordsInDoc;
                        }
                        else
                        {
                            int currentTermID = wordTermIDs[word];
                            double currentTermFreq =(double) termFreqInDoc[currentTermID] / totalWordsInDoc;
                            double neighborProb = lastTermFreq*currentTermFreq/sentenceCoverage;
                            if(neighborProbs.ContainsKey(lastNonStopTermID))
                            {
                                Dictionary<int, double> toTermProbs = neighborProbs[lastNonStopTermID];
                                Dictionary<int, int> toTermFreqs = neighborFreqs[lastNonStopTermID];
                                if(toTermProbs.ContainsKey(currentTermID))
                                {
                                    toTermProbs[currentTermID] += neighborProb;
                                    toTermFreqs[currentTermID] += 1;
                                }
                                else
                                {
                                    toTermFreqs.Add(currentTermID, 1);
                                    toTermProbs.Add(currentTermID, neighborProb);
                                }
                                neighborProbs[lastNonStopTermID] = toTermProbs;
                                neighborFreqs[lastNonStopTermID] = toTermFreqs;
                            }
                            else
                            {
                                Dictionary<int, double> toTermProbs = new Dictionary<int, double>();
                                Dictionary<int, int> toTermFreqs = new Dictionary<int, int>();
                                toTermFreqs.Add(currentTermID, 1);
                                toTermProbs.Add(currentTermID, neighborProb);
                                neighborProbs.Add(lastNonStopTermID, toTermProbs);
                                neighborFreqs.Add(lastNonStopTermID, toTermFreqs);
                            }

                            lastNonStopTermID = currentTermID;
                            lastTermFreq = currentTermFreq;
                        }
                    }
                }
            }
            
            foreach(int leftTermID in neighborFreqs.Keys)
            {
                foreach(int rightTermID in neighborFreqs[leftTermID].Keys)
                {
                    int freq = neighborFreqs[leftTermID][rightTermID];
                    double prob = neighborProbs[leftTermID][rightTermID];
                    double randomProb = (double) termFreqInDoc[leftTermID]*termFreqInDoc[rightTermID]/
                                        (totalWordsInDoc*totalWordsInDoc);
                    if(prob >= Math.Min(1.0, 1.5 * randomProb))
                    {
                        this.AddNeighbor(cmd, leftTermID, rightTermID, freq);
                    }
                }
            }
        }

        private void AddNeighbor(SqlCommand cmd, 
            int leftTermID, int rightTermID, int neighborOccInDoc)
        {
            bool exists = false;
            cmd.CommandType = CommandType.Text;
            cmd.CommandText = string.Format("select count (*) from [{0}] where [{1}]={3} and [{2}]={4}",
                                            instruction.NeighborTable, instruction.LeftTermIDField,
                                            instruction.RightTermIDField,
                                            leftTermID, rightTermID);
            object result = cmd.ExecuteScalar();
            if(result !=DBNull.Value)
            {
                int count = 0;
                bool canParse = int.TryParse(result.ToString(), out count);
                if(canParse && count>0)
                {
                    exists = true;
                }
            }
            if(exists)
            {
                string sql = "update [{0}] set [{3}]=[{3}]+1, [{4}]=[{4}]+{7} where " +
                             "[{1}]={5} and [{2}]={6}";
                cmd.CommandText = string.Format(sql, instruction.NeighborTable,
                                                instruction.LeftTermIDField, instruction.RightTermIDField,
                                                instruction.DocContainingNeighborField,
                                                instruction.NeighborTotalOccuranceField,
                                                leftTermID, rightTermID, neighborOccInDoc);
                cmd.ExecuteNonQuery();
            }
            else
            {
                string sql = "insert into [{0}] ([{1}],[{2}],[{3}],[{4}]) values ({5},{6},1,{7})";
                cmd.CommandText = string.Format(sql, instruction.NeighborTable,
                                    instruction.LeftTermIDField, instruction.RightTermIDField,
                                    instruction.DocContainingNeighborField, instruction.NeighborTotalOccuranceField,
                                    leftTermID, rightTermID, neighborOccInDoc);
                cmd.ExecuteNonQuery();
            }
        }
        #endregion

        #region trend 
        private void AddTermToTrend(SqlCommand cmd, int year,
            Dictionary<int,int> termFreqs)
        {
            if(year < 1900 || year > DateTime.Now.Year)
                return;
            cmd.CommandType = CommandType.Text;
            string existSql = "select count (*) from [{0}] where [{1}]={3} and [{2}]={4}";
            foreach(int termID in termFreqs.Keys)
            {
                cmd.CommandText = string.Format(existSql, instruction.TrendTable, instruction.TermIDField3,
                                                instruction.TermYearField,
                                                termID, year);
                object existObj = cmd.ExecuteScalar();
                int count = 0;
                bool canParse = int.TryParse(existObj.ToString(), out count);
                if(canParse && count>0)
                {
                    string sql = "update [{0}] set [{3}]=[{3}]+{7}, [{4}]=[{4}]+1 where " +
                                 "[{1}]={5} and [{2}]={6}";
                    cmd.CommandText = string.Format(sql, instruction.TrendTable, instruction.TermIDField3,
                                        instruction.TermYearField, instruction.SumTermFreqField,
                                        instruction.SumDocFreqField, termID, year, termFreqs[termID]);
                    cmd.ExecuteNonQuery();
                }
                else
                {
                    string sql = "insert into [{0}] ([{1}],[{2}],[{3}],[{4}]) values ({5},{6},{7},1)";
                    cmd.CommandText = string.Format(sql, instruction.TrendTable,
                                                    instruction.TermIDField3, instruction.TermYearField,
                                                    instruction.SumTermFreqField, instruction.SumDocFreqField,
                                                    termID, year, termFreqs[termID]);
                    cmd.ExecuteNonQuery();
                }
            }
        }
        #endregion
    }
}
